#include "ThreadPool.h"
#include "Task.h"
#include "Log.h"

#include <time.h>
#include <sys/time.h>
#include <algorithm>
#include <boost/bind.hpp>

namespace sevent{

ThreadPool::ThreadPool(): _started(false){
	_taskSize = 0;
	pthread_cond_init(&_cond, NULL);
	pthread_mutex_init(&_mutex, NULL);
}

ThreadPool::ThreadPool(const std::string &name):_name(name), _started(false){
	_taskSize = 0;
	pthread_cond_init(&_cond, NULL);
	pthread_mutex_init(&_mutex, NULL);
}

ThreadPool::~ThreadPool(){
	pthread_cond_destroy(&_cond);
	pthread_mutex_destroy(&_mutex);
}

void ThreadPool::clean(){
	if (_started) {
		stop();
	}
	//delete all thread, task
}

void ThreadPool::start(int number){
	assert(!_started);
	assert(_threads.empty());
	_started = true;
	_threadsNum = number;
	LOG_INFO_VA("thread pool %s started with number: %d", _name.c_str(), _threadsNum);
	char buf[16] = {0};
	for (int i = 0; i < _threadsNum; ++i) {
		snprintf(buf, 15, "_%d", i);
		Thread *t = new Thread(boost::bind(&ThreadPool::threadInit, this), boost::bind(&ThreadPool::threadRun, this), boost::bind(&ThreadPool::threadStop, this));
		_threads.push_back(t);
		t->start();
	}
	return;
}

int ThreadPool::wait(){
	for(std::vector<Thread *>::iterator iter = _threads.begin(); iter != _threads.end(); ++iter) {
		(*iter)->wait();
	}
	return 0;
}

int ThreadPool::stop(){
	_started = false;
	pthread_cond_broadcast(&_cond);
//	for_each(_threads.begin(), _threads.end(),[](Thread * t) {t->join();});
	for(std::vector<Thread *>::iterator iter = _threads.begin(); iter != _threads.end(); ++iter) {
		(*iter)->stop();
	}
	return 0;
}

int ThreadPool::addTask(Task * t){
	
	int ret = 0;
	pthread_mutex_lock(&_mutex);
//	if(_taskSize > _threadsNum*10) {
//		LOG_WARN_VA("task queue size exceed %d. Reject.", _threadsNum*10);
//		pthread_mutex_unlock(&_mutex);
//		return -1;
//	}
	_tasks.push_back(t);
	LOG_DEBUG_VA("current task size: %d", _taskSize);
	_taskSize++;
	pthread_mutex_unlock(&_mutex);
	pthread_cond_signal(&_cond);
	return ret;
}

Task *ThreadPool::getNextTask(){
	pthread_mutex_lock(&_mutex);
	while(_tasks.empty() && _started) {
		pthread_cond_wait(&_cond, &_mutex);
	}
	Task *t = NULL;
	if(!_tasks.empty()) {
		t = _tasks.front();
		_tasks.pop_front();
		_taskSize--;
	}
	pthread_mutex_unlock(&_mutex);
	return t;
}

int ThreadPool::threadInit(){
	return 0;
}

int ThreadPool::threadRun(){
	
	while (_started) {
		Task *t = getNextTask();
		if (!t) {
			LOG_ERROR_VA("got null task");
			continue;
		}
		int ret = t->execute();
		if (ret < 0) {
			LOG_ERROR_VA("task execution error");
		}
		delete t;
	}

	LOG_INFO_VA("ThreadPool stoped");
	return 0;
}

int ThreadPool::threadStop(){
	return 0;
}

}
